package com.ebs.md.fxmonitor.client.impl;


import java.io.File;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.log4j.Logger;
import org.springframework.core.io.ClassPathResource;

import com.cameronsystems.fix.exception.MessageException;
import com.ebs.commons.comp.IParsedPayload;
import com.ebs.commons.comp.IPubSubChannel;
import com.ebs.commons.comp.ParsedPayload;
import com.ebs.commons.db.DataSourceException;
import com.ebs.commons.db.SPLoader;
import com.ebs.commons.logger.MessageText;
import com.ebs.commons.threads.EbsThreadFactory;
import com.ebs.commons.translators.fix_translator.api.ICAPFixConstants;
import com.ebs.commons.translators.fix_translator.impl.FixTranslatorImpl;
import com.ebs.md.fxmonitor.client.api.ClientMessageListener;
import com.ebs.md.fxmonitor.client.api.NetlinkFixConstants;
import com.ebs.md.fxmonitor.topics.ClientTopicKeys;
import com.ebs.md.fxmonitor.topics.JmxTopicKeys;
import com.ebs.md.fxmonitor.topics.Keys;

/**
 * <p>Title: Netlink Project</p>
 *
 * <p>Description: Netlink Component</p>
 *
 * <p>Copyright: Copyright (c) 2007</p>
 *
 * <p>Company: ICAP</p> 
 * 
 * <p>$Id: //depot/MarketData/fxmonitor/main/src/main/java/com/ebs/md/fxmonitor/client/impl/ClientMediator.java#88 $ </p>
 * <p>$DateTime: 2012/08/16 12:17:24 $ </p>
 * <p>$Revision: #88 $ </p>
 * <p>$Change: 113278 $ </p>
 * <p>$Author: Arik.Sher $ </p>
 */
public class ClientMediator implements ClientMessageListener, MessageListener 
{
    private boolean m_wasEverSubscribed; //if client was never subscribed before, logon message must have the reset flag
    enum State { Connected, LoggedIn, Subscribed, SubscribedAndDisconnected, LoggedOut} ;
    
    final int EVENT_DOWN = 256;
    final int EVENT_CONNECTED = 257;
    final int EVENT_SIGNED_ON = 258;
    final int EVENT_RECOVERY_REQUESTED = 259;
    final int EVENT_RECOVERING = 260;
    final int EVENT_RUNNING = 261;
    final int EVENT_ARB_UP_LIVE = 1;
    final int EVENT_ARB_DOWN = 2;
    final int EVENT_ARB_UP_MOCK = 13;

    
    static final int PIP_TYPE_WHOLE = 1;
    
    static final int TRADE_TYPE_SPOT = 0;
    static final int TRADE_TYPE_FIX  = 1;
    static final int TRADE_TYPE_CM   = 100;

    static final int MT_LIMIT   = 0; // Core.CODE_MatchingType_LIMIT
    static final int MT_FIX     = 1; // Core.CODE_MatchingType_FIX
    static final int MT_CM      = 2; // Core.CODE_MatchingType_CROSSING
    static final int MT_NDF_SL  = 3; // Core.CODE_MatchingType_NDF_SINGLE
    static final int MT_NDF_ML  = 4; // Core.CODE_MatchingType_NDF_MULTI
    static final int MT_SWAPS   = 5; // Core.CODE_MatchingType_SWAPS
    
        private State m_state;
        private FixInProcessSession m_fixSession;
        private IPubSubChannel m_NdsChannel;
        private Session m_jmsSession;
//        private Translator m_binfilesTranslator;
        private FixTranslatorImpl m_fixTranslator;
        private int localSessionSequencer;

        private String m_region;
        private String m_side;
        int m_currentSubscriptionId;
        private int m_disconnectEventCounter = 1;
        final static Logger s_logger = Logger.getLogger(ClientMediator.class);
        
        private int m_bufferedMessagesCounter;

        private int m_maxMessagesToBuffer = 100000; //max number of messages we allow FIX engine to buffer before closing the session
        
        private Map<String, Map<String,Object>> m_name_to_currency = new HashMap<String, Map<String,Object>>();
        
        private Map<String, Map<String,Object>> m_pairs = new HashMap<String, Map<String,Object>>();
        
        private Map<String, String> m_pair_name_to_id = new HashMap<String, String>();
        
        private Map<String, String> m_pair_names = new HashMap<String, String>();
        
        private Map<String, List<String>> m_msg_distributor = new HashMap<String, List<String>>();;
        
        private Map<String,String> m_fix_names = new HashMap<String,String>();
        
        private Map<String, String> m_snapshot_filter = new HashMap<String, String>();
        
        private SPLoader m_spLoader;

        private Object m_monitor = new Object();
        
        private boolean m_testSubscription = false;
        
        private ScheduledExecutorService m_scheduledExecutor;
        
        private TestSubscribe m_testSubscribe = new TestSubscribe();
        
        private PeriodicRecovery m_PeriodicRecovery = new PeriodicRecovery();
        
        public ClientMediator(
        		IPubSubChannel ndsChannel,
        		FixTranslatorImpl fixTranslator,
        		String region,
        		String side,
        		SPLoader spLoader) 
        {
            m_state = State.LoggedOut;
            s_logger.info("ClientMediator moved to the LoggedOut state");
            m_wasEverSubscribed = false;
            m_NdsChannel = ndsChannel;  
            m_NdsChannel.connectListeners();
           	m_fixTranslator = fixTranslator;
           	m_region = region;
           	m_side = side;
           	startFixSession();
            s_logger.info("ClientMediator is created");
            m_spLoader = spLoader;
        }
        

        public void setSpLoader(SPLoader spLoader)
        {
        	m_spLoader = spLoader;
        }
        
        public synchronized void startFixSession()
        {
        	try
        	{
        		ClassPathResource ccr = new ClassPathResource("nl_config.xml");
        		File cameronConfigFile = ccr.getFile();
        		m_fixSession = new FixInProcessSession();
        	
        		m_fixSession.setCameronConfig(cameronConfigFile);
        		m_fixSession.setClientMessageListener(this);
        		m_fixSession.start();
        		s_logger.info("ClientMediator m_fixSession:" + m_fixSession);        		
        	}
        	catch (Exception e)
        	{
        		s_logger.error("ClientMediator got exception:",e);
        	}
        	
        }
        
    	private void startTestSubscribe()
    	{    	
    		s_logger.info("ClientMediator scheduled TestSubscribe in 2 seconds");
        	m_scheduledExecutor.schedule(m_testSubscribe, 2, TimeUnit.SECONDS);    	
    	}        
        
        public void setTestSubscription(boolean testSubscription)
        {
        	m_testSubscription = testSubscription;
        }


        
        public void setMessageDistribution(Map<String, List<String>> msg_distributor)
        {
        	m_msg_distributor = msg_distributor;
        }


        public void setMaxMessagesToBuffer(int limit){
            m_maxMessagesToBuffer = limit;
        }


        public void setRegion(String region){
            m_region = region;
        }

        public void setSide(String side){
            m_side = side;
        }

        private boolean isSessionOpen(){
            return State.Subscribed == m_state || State.SubscribedAndDisconnected == m_state;
        }

        public void setFixSession(FixInProcessSession sender) {
            m_fixSession = sender;
        }

//        public void setDestinationMap(DestinationMap map) {
//            m_destinations = map;
//        }

        public void setJmsSession(Session jms) {
            m_jmsSession = jms;
        }

        public void setFixTranslator(FixTranslatorImpl trans) {
            m_fixTranslator = trans;
        }


        public void init() {
            if (null == m_spLoader) {
                throw new RuntimeException("SPLoader must be defined");
            }
            if (null == m_fixSession) {
                throw new RuntimeException("Fix session must be defined");
            }
            if (null == m_fixTranslator) {
                throw new RuntimeException("Fix Translator must be defined");
            }
            if (null == m_region) {
                throw new RuntimeException("Region must be defined");
            }
            if (null == m_side) {
                throw new RuntimeException("Side must be defined");
            }
            
        	m_scheduledExecutor = Executors.newSingleThreadScheduledExecutor(new EbsThreadFactory("ClientMediator_scheduler"));
        	
        	refreshMaps();
            
            if(m_testSubscription)
            {
            	s_logger.info("ClientMediator starting TestSubscribe thread");            	
            	startTestSubscribe();
            }
                        
            s_logger.info("ClientMediator init is done");
        }
        
        private void populateCurrencyMap()
        {
            	try
            	{
            		Map<String,Object> params = new HashMap<String,Object>();
            		params.put(m_spLoader.getSPLabel(), "GET_CURRENCY_BIN");
            		params.put("IS_DELETED", 0);
            		params.put("IS_TEST", 0);
            		List<Map<String,Object>> result = m_spLoader.getRecordSet(params); 
            		s_logger.info("ClentMediator result for GET_CURRENCY_BIN:" + result);
            		if((result == null) || (result.size() <= 0))
            		{
            			return;
            		}
            		Map<String,Map<String,Object>> name_to_currency = new HashMap<String,Map<String,Object>>();
                	for(Map<String,Object> nextCurrency : result)
                	{
                		if(nextCurrency.containsKey("ccy_name"))
                		{
                			String isocode = (String)nextCurrency.get("ccy_name");
                			name_to_currency.put(isocode, nextCurrency);
                			s_logger.info("ClientMediator added to collection currency name:" + isocode);
                		}
                	}
                	
                	for(Map<String,Object> nextCurrency : result)
                	{
                		Object alias_object = nextCurrency.get("ccy_alias");
                		
                		String alias = null;
                		if(alias_object != null)
                		{
                			alias = (String)alias_object;
                		}
                		if((alias != null)&&(alias.length() > 0)&&(!name_to_currency.containsKey(alias)))
                		{
                			name_to_currency.put(alias, nextCurrency);
                			s_logger.info("ClientMediator added to collection alias currency name:" + alias);
                		}                		
                	} 
                	m_name_to_currency = name_to_currency;
            	}
            	catch(DataSourceException de)
            	{
            		s_logger.error("Failed to get currencies from RDB", de);
            	}
            	
        }
        
        
        private void populatePairMap()
        {              	
        	try
        	{
        		Map<String,Object> params = new HashMap<String,Object>();
        		params.put(m_spLoader.getSPLabel(), "GET_PAIRS_BIN");
        		params.put("IS_MKT_ACTIVE", 1);
        		s_logger.info("ClentMediator params for GET_PAIRS_BIN:" + params);
        		List<Map<String,Object>> pairsSet = m_spLoader.getRecordSet(params);
        		s_logger.info("ClentMediator result for GET_PAIRS_BIN:" + pairsSet);
        		if((pairsSet == null) || (pairsSet.size() <= 0))
        		{
        			return;
        		}        		   
        		Map<String, Map<String,Object>> pairs = new HashMap<String, Map<String,Object>>();
        		Map<String, String> pair_names = new HashMap<String, String>();
        		Map<String, String> pair_name_to_id = new HashMap<String, String>();
            	for(Map<String,Object> nextPair : pairsSet)
            	{
            		if(nextPair.containsKey("pair_id"))
            		{
            			String basename = (String)nextPair.get("base_ccy_name");
            			nextPair.put("base_ccy_object", m_name_to_currency.get(basename));
            			String localname = (String)nextPair.get("local_ccy_name");
            			nextPair.put("local_ccy_object", m_name_to_currency.get(localname));
            			String pairid = "" + (Integer)nextPair.get("pair_id");
            			pairs.put(pairid, nextPair);
            			String short_alias_name = (String)nextPair.get("short_alias_name");
            			pair_names.put(basename + localname, short_alias_name);
            			pair_name_to_id.put(basename + localname, pairid);
            			s_logger.info("ClientMediator added to collection pair:" + basename + "/" + localname);
            		}
            		else
            		{
            			s_logger.error("ClientMediator pair does not have pair_id key. nextPair:" + nextPair);
            		}
            	}  
            	m_pairs = pairs;
            	m_pair_names = pair_names;
            	m_pair_name_to_id = pair_name_to_id;
        	}
        	catch(DataSourceException de)
        	{
        		s_logger.error("Failed to get currencies from RDB", de);
        	}
        }
        
        private void refreshMaps()
        {
        	synchronized(m_monitor)
        	{
        		populateCurrencyMap();
        		populatePairMap(); 
        		populateFixList();
            	m_scheduledExecutor.schedule(m_PeriodicRecovery, 2, TimeUnit.HOURS);        		
        	}
        }

        
        private void populateFixList()
        {      
        	Map<String, String> fix_names = new HashMap<String, String>();
        	try
        	{
        		Map<String,Object> params = new HashMap<String,Object>();
        		params.put(m_spLoader.getSPLabel(), "GET_FIX_NAMES");
        		List<Map<String,Object>> pairsSet = m_spLoader.getRecordSet(params);        		
        		s_logger.info("ClentMediator result for GET_FIX_NAMES:" + pairsSet);
        		if((pairsSet == null) || (pairsSet.size() <= 0))
        		{
        			return;
        		}
            	for(Map<String,Object> nextFix : pairsSet)
            	{
            			String fix_id = nextFix.get("fix_id").toString();
            			String fix_name = nextFix.get("fix_name").toString();
            			String fix_source = nextFix.get("fix_source").toString();
            			String fix_time = nextFix.get("fix_time").toString();
            			String fix_time_zone = nextFix.get("fix_time_zone").toString();
            			fix_names.put(fix_id, fix_name);
            			s_logger.info("ClientMediator added to collection fixing fix_id:" + fix_id +
            					" fix_name:" + fix_name );
            	} 
            	m_fix_names = fix_names;
        	}
        	catch(DataSourceException de)
        	{
        		s_logger.error("Failed to get list of fixings from RDB", de);
        	}
        }

        
        
        /**
         * Called when everything is ready to receive messages
         */
        public void onServerInitialized(){
            //set logout request to persister
            sendDisconnectMessageToRouter("startup clean up", true, true);
        }

        /**
         * onClientConnected
         *
         * @param ip String
         */
        public void onClientConnected(String ip) {
            s_logger.info(new MessageText("TDL070011", ip));
            synchronized(m_monitor){
                if (m_state == State.LoggedOut) {
                    m_state = State.Connected;
                }
            }
        }

        /**
         * onClientDisconnected
         *
         * @param reason String
         */
        public void onClientDisconnected(String reason) {
            s_logger.info(new MessageText("NETLINK_0100016", m_fixSession.getClientIp(), reason));

            synchronized(m_monitor){
                if (m_state != State.LoggedOut) {
                    sendDisconnectMessageToRouter(reason, false, false);
                }

                if (isSessionOpen()) {
                    m_state = State.SubscribedAndDisconnected;
                }
                else {
                    m_state = State.LoggedOut;
                }
            }
        }

//        private void addRegionSide(IParsedPayload payload){
//            payload.addField(ClientTopicKeys.GlobalRegion, m_region);
//            payload.addField(ClientTopicKeys.GlobalSide, m_side);
//        }

        private void sendDisconnectMessageToRouter(String reason, boolean gracefull, boolean disconnectAll) {
            IParsedPayload payload = new ParsedPayload();
            payload.addField("type", -3);
            String clientIp = disconnectAll ? "all clients" : m_fixSession.getClientIp();
            if(!disconnectAll){
                payload.addField(ClientTopicKeys.ClientIp, clientIp);
            }
            payload.addField(ClientTopicKeys.Event, 1);
            String disconnectReason = reason;//gracefull ? "Gracefull logout" : reason;
            s_logger.info(new MessageText("TDL070009", clientIp, disconnectReason));
            payload.addField(ClientTopicKeys.Reason, disconnectReason);
            payload.addField(ClientTopicKeys.Gracefull, gracefull);
//            addRegionSide(payload);
            s_logger.info(new MessageText("NETLINK_0100022", Keys.MessageTypes.client_disconnect, payload.getParsedMessage().toString()));
//            m_destinations.sendMessage(Keys.MessageTypes.client_disconnect, payload);
            submitInternalMessage("client_connection_event",payload);
        }

        /**
         * onClientException
         *
         * @param ex Exception
         */
        public void onClientException(Exception ex) {
            s_logger.error(new MessageText("TDL070013", m_fixSession.getClientIp()), ex);
        }

        /**
         * onClientLogin
         *
         * @param payload IParsedPayload
         * @throws MessageException
         */
        public void onClientLogin(IParsedPayload payload) {
        	        	
        	if(payload == null)
        	{
        		payload = new ParsedPayload();
        		s_logger.error("No data received onClientLogin");
        	}
        	
        	if (m_fixSession == null)
        	{
        		startFixSession();
        	}
        	
            s_logger.info(new MessageText("NETLINK_0100017", m_fixSession.getClientIp(), payload.getParsedMessage().toString()));

            int resetFlag = payload.getParsedIntField(ClientTopicKeys.ResetSequenceNumberFlag);

            synchronized(m_monitor){

                if (0 == resetFlag && m_state != State.SubscribedAndDisconnected) {
                    forceClientLogout("NETLINK_0100003", null, false);
                    return;
                }
                if (0 == resetFlag && (m_state == State.Subscribed || m_state == State.LoggedIn)) {//undeterministic state
                    forceClientLogout("NETLINK_0100024", null, true);
                    return;
                }

                if (isSessionOpen() && 1 == resetFlag) {
                    sendUnsubscribeMessageToRouter();
                }

                if(m_state == State.SubscribedAndDisconnected && 0 == resetFlag){
                    m_bufferedMessagesCounter = 0;
                    m_state = State.Subscribed;
                }else{
                    m_state = State.LoggedIn;
                }
            }

            payload.addField(ClientTopicKeys.Event, 0); //connect
//            addRegionSide(payload);
            s_logger.info(new MessageText("NETLINK_0100022", Keys.MessageTypes.client_connect, payload.getParsedMessage().toString()));
//            m_destinations.sendMessage(Keys.MessageTypes.client_connect, payload);
            submitInternalMessage("client_connection_event",payload);
            onClientSubscribe(payload);
        }


        /**
         * CAlled when client logs out gracefully
         * @param reason String - reason for disconnect
         * @param gracefull boolean - true if a gracefull disconnect
         */
        public void onClientLogout(String reason, boolean gracefull) {        	
            s_logger.info(new MessageText("NETLINK_0100018", m_fixSession.getClientIp(), reason, gracefull));

            synchronized(m_monitor){
                if (isSessionOpen()) { //send unsubscribe message
                    sendUnsubscribeMessageToRouter();
                }
                if (m_state != State.LoggedOut) {
                    sendDisconnectMessageToRouter(reason, gracefull, false);
                }
                m_state = State.LoggedOut;
            }
        }

        private void sendUnsubscribeMessageToRouter(){        	
            IParsedPayload message = new ParsedPayload();
            //type -1 is unsubscribe to Sync
            message.addField("type", -4);
            message.addField(ClientTopicKeys.SubscriptionId, m_currentSubscriptionId);
            message.addField(ClientTopicKeys.SubscribeType, 0);
//            addRegionSide(message);
            s_logger.info(new MessageText("NETLINK_0100022", Keys.MessageTypes.subscription, message.getParsedMessage().toString()));
//            m_destinations.sendMessage(Keys.MessageTypes.subscription, message);
            submitInternalMessage("subscription",message);
        }

        /**
         * onClientSubscribe
         *
         * @param inMessage IParsedPayload
         */
        public void onClientSubscribe(IParsedPayload inMessage) {
            s_logger.info(new MessageText("NETLINK_0100019", m_fixSession.getClientIp(), inMessage.getParsedMessage().toString()));

            synchronized(m_monitor){
                if (isSessionOpen()) { //duplicate subscription
                    forceClientLogout("NETLINK_0100002", "Can only subscribe once", true);
                    return;
                }
                if (m_state != State.LoggedIn) {
                    forceClientLogout("NETLINK_0100027", null, true);
                    return;
                }

            }

            m_snapshot_filter = new HashMap<String, String>();
            m_currentSubscriptionId = sendSubscriptionToDb(inMessage);
            IParsedPayload subscrResponsePayload = new ParsedPayload();
            buildSubscrResponsePayload(subscrResponsePayload);

            //send to fix session
            sendFixMessage(subscrResponsePayload, ICAPFixConstants.SubtypeSubscriptionResponse, null);
                inMessage.addField("type", -4);
                inMessage.addField(ClientTopicKeys.SubscriptionId, m_currentSubscriptionId);
                inMessage.addField(ClientTopicKeys.SubscribeType, 1);
                s_logger.info(new MessageText("NETLINK_0100022", Keys.MessageTypes.subscription, inMessage.getParsedMessage().toString()));
//                m_destinations.sendMessage(Keys.MessageTypes.subscription, inMessage);
                submitInternalMessage("subscription",inMessage);
                synchronized(m_monitor){
                    m_bufferedMessagesCounter = 0;
                    m_state = State.Subscribed;
                    m_wasEverSubscribed = true;
                }
//            }
        }
        
        private void buildSubscrResponsePayload(IParsedPayload result)
        {
          synchronized(m_monitor)
          {        	
        	result.addField(ClientTopicKeys.MessageType, ICAPFixConstants.SubtypeSubscriptionResponse);
        	result.addField(ClientTopicKeys.MessageVersion, "1");
        	result.addField(ClientTopicKeys.SubscriptionStatus, "0");  
        	result.addField("mq_msg_type","M0002");
        	if((m_name_to_currency == null)||(m_name_to_currency.size() <= 0))
        	{
        		if(m_name_to_currency == null)
        		{
        			s_logger.info("ClientMediator m_name_to_currency == null going to populate currencies");
        		}
        		else if(m_name_to_currency.size() <= 0)
        		{
        			s_logger.info("ClientMediator m_name_to_currency.size():" + m_name_to_currency.size() + " going to populate currencies");
        		}
        		populateCurrencyMap();
        	}
        	if((m_pairs == null)||(m_pairs.size() <= 0))
        	{
        		s_logger.info("ClientMediator m_pairs is null or size is 0. Going to populate pairs");
        		populatePairMap();
        	}
            List<Map<String,Object>> ccyPairs = new ArrayList<Map<String,Object>>();
            for (String pairKey : m_pairs.keySet())
            {            	
            	Map<String, Object>ccyPair = m_pairs.get(pairKey);
            	String pairIsoName = (String)ccyPair.get("short_iso_name");
            	String pairName = (String)ccyPair.get("short_alias_name");
            	String baseName = (String)ccyPair.get("base_ccy_name");
            	if(!m_name_to_currency.containsKey(baseName))
            	{
	                s_logger.error("Incorrect base ccy: " + baseName);
	          		continue;
	           	}
            	           	
            	Map<String,Object> baseCcy = m_name_to_currency.get(baseName);
            	
            	Map<String,Object> ccyPairDetails = new HashMap<String,Object>();
            	
            	int pairID = Integer.parseInt(pairKey);
            	
            	int adjPip = (Integer)ccyPair.get("pip_type");             	
            	ccyPairDetails.put(ClientTopicKeys.CcyPair, pairName);
            	//trade_type=0 is spot regular
            	ccyPairDetails.put("trade_type", TRADE_TYPE_SPOT);
            	String instrument = (String)ccyPair.get("instrument");
            	ccyPairDetails.put(ClientTopicKeys.Instrument, instrument);
            	ccyPairDetails.put(ClientTopicKeys.Tenor, (String)ccyPair.get("tenor"));
            	ccyPairDetails.put("near_tenor", (String)ccyPair.get("near_tenor"));
            	ccyPairDetails.put(ClientTopicKeys.CcyPairScaleFactor, (Integer)ccyPair.get("scale_factor"));
            	String xpips = (String)ccyPair.get("xpips"); 
            	ccyPairDetails.put(ClientTopicKeys.XPips, xpips);
            	ccyPairDetails.put(ClientTopicKeys.IncrementType, (String)ccyPair.get("increment_type"));
            	ccyPairDetails.put(ClientTopicKeys.BigFigureSize, (Integer)ccyPair.get("big_figure_size"));            	
            	ccyPairDetails.put(ClientTopicKeys.PricePrecision, (Integer)ccyPair.get("pips_end_position"));
            	String testLotSize = (String)ccyPair.get("lot_size");
            	if((testLotSize == null) || (testLotSize.length()<=0))
            	{
            		testLotSize = "0";
            	}
            	ccyPairDetails.put(ClientTopicKeys.LotSize, testLotSize);
            	
            	if(!((instrument.equalsIgnoreCase("NDF_SL"))||(instrument.equalsIgnoreCase("NDF_ML"))||(instrument.equalsIgnoreCase("FWD"))))
            	{
        			addNlogPair(ccyPairs, ccyPairDetails, TRADE_TYPE_SPOT, "regular");
            	}


            	if(instrument.equalsIgnoreCase("SPOT"))
            	{                	
                	Map<String,Object> fixCcyPairDetails = new HashMap<String,Object>(ccyPairDetails);
                	if(adjPip == PIP_TYPE_WHOLE) {
                		modifyPrecisionForFIX(ccyPair, fixCcyPairDetails);
                	}
                	//trade_type=1 is FIXing
                	addNlogPair(ccyPairs, fixCcyPairDetails, TRADE_TYPE_FIX, "fix");
                	          
                	Map<String,Object> cmCcyPairDetails = new HashMap<String,Object>(ccyPairDetails);            	
               		if(adjPip == PIP_TYPE_WHOLE) {
               			modifyPrecisionForFIX(ccyPair, cmCcyPairDetails);
            		}
                	addNlogPair(ccyPairs, cmCcyPairDetails, TRADE_TYPE_CM, "cm");
            	}


            	if((instrument.equalsIgnoreCase("NDF_SL"))||(instrument.equalsIgnoreCase("NDF_ML")))
            	{
            		Map<String,Object> swapCcyPairDetails = new HashMap<String,Object>(ccyPairDetails);
            		modifyPrecisionForFIX(ccyPair, swapCcyPairDetails);            		
            		//trade_type=1 for NDF Swaps
                	addNlogPair(ccyPairs, swapCcyPairDetails, TRADE_TYPE_FIX, "nds");
            	}


            	if(instrument.equalsIgnoreCase("FWD"))
            	{
            		Map<String,Object> fwdCcyPairDetails = new HashMap<String,Object>(ccyPairDetails);
            		modifyPrecisionForFIX(ccyPair, fwdCcyPairDetails);            		
            		//trade_type=1 for NDF Swaps
            		addNlogPair(ccyPairs, fwdCcyPairDetails, TRADE_TYPE_FIX, "fwd");
            	}
            	
            } // end loop
            
        	result.addField(ClientTopicKeys.NumberOfCcyPairs, ccyPairs);
          } // end synch
          
        }


		protected void modifyPrecisionForFIX(Map<String, Object> srcCcyPair,	Map<String, Object> destCcyPairDetails) {
			destCcyPairDetails.put(ClientTopicKeys.IncrementType,  /*(String) */ srcCcyPair.get("increment_type_fix"));
			destCcyPairDetails.put(ClientTopicKeys.BigFigureSize,  /*(Integer)*/ srcCcyPair.get("big_figure_size_fix"));
			destCcyPairDetails.put(ClientTopicKeys.PricePrecision, /*(Integer)*/ srcCcyPair.get("pips_end_position_fix"));
			destCcyPairDetails.put(ClientTopicKeys.XPips,          /*(String) */ srcCcyPair.get("xpips_fix"));
		}


		/**
		 * Add and Log pair.
		 * 
		 * @param ccyPairs
		 * @param ccyPairDetails
		 * @param trade_type
		 * @param logicalName - name for display
		 */
		protected void addNlogPair(List<Map<String, Object>> ccyPairs, Map<String, Object> ccyPairDetails, int trade_type, String logicalName) 
		{
        	ccyPairDetails.put("trade_type", trade_type);
        	
			ccyPairs.add(ccyPairDetails);

        	String pairName   = (String)ccyPairDetails.get("short_alias_name");
        	String instrument = (String)ccyPairDetails.get("instrument");

			s_logger.info("Currency pair: '" + pairName + "' , trade type:" + trade_type +  " instrument:" + instrument + " is added to the response. '" + logicalName +  "' ccyPairDetails:" + ccyPairDetails);
		}
        
        protected <T> T nvl(Map<String, Object> values, String entryName, T defaultValue) {
        	Object val = values.get(entryName);
        	
        	if ( val != null ) {
        		return (T)val;
        	} else {
        		return defaultValue;
        	}
        }
        
        protected int makeMinimumMessageVersion(IParsedPayload payload, int minVersion) {
        	int currVersion = payload.getParsedIntField(ClientTopicKeys.MessageVersion, (minVersion-1));
        	
        	if ( currVersion < minVersion ) {
        		currVersion = minVersion;
        		payload.addField(ClientTopicKeys.MessageVersion, minVersion);
        	}
        	return currVersion;
        }
        
        /**
         * adding as second size only if value of it is greater than 0.(means it exists)
         * 
         * @param payload
         */
        protected void addTotalAmountAsSecondSize(IParsedPayload payload) {
        	long amountValue = payload.getParsedLongField("qte_amt_result", -1l);
        	
        	if ( amountValue > 0 ) {
        		addSecondSize(payload, amountValue, NetlinkFixConstants.MDSecSizeType_Total);
        	}
        }
        protected void addTotalAmountAsSecondSize(Map<String,Object> valuesMap) {
        	long amountValue = nvl(valuesMap, "qte_amt_result", -1l);
        	
        	if ( amountValue > 0 ) {
        		addSecondSize(valuesMap, amountValue, NetlinkFixConstants.MDSecSizeType_Total);
        	}
        }
        /**
         * adding as second size as is (negative values allowed)
         * 
         * @param payload
         */
        protected void addDeltaAmountAsSecondSize(IParsedPayload payload) {
        	long amountValue = payload.getParsedLongField("qte_amt_result", 0l);
        	
       		addSecondSize(payload, amountValue, NetlinkFixConstants.MDSecSizeType_Delta);
        }
        protected void addDeltaAmountAsSecondSize(Map<String,Object> valuesMap) {
        	long amountValue = nvl(valuesMap, "qte_amt_result", 0l);
        	
       		addSecondSize(valuesMap, amountValue, NetlinkFixConstants.MDSecSizeType_Delta);
        }
        
        protected void addSecondSize(IParsedPayload payload, long secSizeValue, int secSizeType) {
        	List<Map<String,Object>> secSizeListOfMap;
        	// make sure List of mapS exists on the payload
        	Object parsedField = payload.getParsedField("fix_second_sizes");
        	if ( parsedField != null ) {
        		secSizeListOfMap = (List<Map<String,Object>>)parsedField;
        	} else {
        		secSizeListOfMap = new ArrayList<Map<String,Object>>(2);
        		payload.addField("fix_second_sizes", secSizeListOfMap);
        	}
        	Map<String,Object> secSizeMap = new HashMap<String, Object>(2);
        	
        	secSizeMap.put("fix_second_size_type",  secSizeType);
        	secSizeMap.put("fix_second_size_value", secSizeValue);
        	
        	secSizeListOfMap.add(secSizeMap);
        }
        
        protected void addSecondSize(Map<String,Object> valuesMap, long secSizeValue, int secSizeType) {
        	List<Map<String,Object>> secSizeListOfMap;
        	// make sure List of mapS exists on the valuesMap
        	Object parsedField = valuesMap.get("fix_second_sizes");
        	if ( parsedField != null ) {
        		secSizeListOfMap = (List<Map<String,Object>>)parsedField;
        	} else {
        		secSizeListOfMap = new ArrayList<Map<String,Object>>(2);
        		valuesMap.put("fix_second_sizes", secSizeListOfMap);
        	}
        	Map<String,Object> secSizeMap = new HashMap<String, Object>(2);
        	
        	secSizeMap.put("fix_second_size_type",  secSizeType);
        	secSizeMap.put("fix_second_size_value", secSizeValue);
        	
        	secSizeListOfMap.add(secSizeMap);
        }
        
        private int sendSubscriptionToDb(IParsedPayload payload) {
            return ++localSessionSequencer;
        }

        private void sendFixMessage(IParsedPayload payload, String subType, IParsedPayload latencyPayload) {
            if((subType != null)&&(subType.length() > 0))
            {
            	payload.addField("mq_msg_type", subType);
            
            }
            sendFixMessage(payload, latencyPayload);
        }

        private void sendFixMessage(IParsedPayload payload, IParsedPayload latencyPayload) {
            com.cameronsystems.fix.message.Message fixMessage = m_fixTranslator.encode(payload.getParsedMessage());
            if((m_testSubscription) && (fixMessage != null))
            {
            	s_logger.info("TestFixMsg:" + new String(fixMessage.getBytes()) + " payload:" + payload.getParsedMessage());
            }
            if(null != fixMessage)
            {//error
                if(null != latencyPayload)
                {
                    payload.addField(ClientTopicKeys.ClientBeforeFix, System.currentTimeMillis());
                }
                
                
                synchronized(m_monitor){
                	String mqType = payload.getParsedStringField("mq_msg_type", "");
                	if(((m_state == State.SubscribedAndDisconnected)||(m_state == State.Subscribed)||(mqType.equalsIgnoreCase("M0002")))&& (!m_testSubscription))
                    {
                    	m_fixSession.send(fixMessage);
                    }
                	else
                	{
                		s_logger.info("ClientMediator Msg Dropped. m_state:" + m_state +
                				" payload:" + payload.getParsedMessage());
                	}
                	
                    if (m_state == State.SubscribedAndDisconnected) {
                        if (m_maxMessagesToBuffer > 0 && ++m_bufferedMessagesCounter > m_maxMessagesToBuffer){
                            MessageText mess = new MessageText("NETLINK_0100025", m_maxMessagesToBuffer);
                            forceClientLogout(mess, null, true);
                        }
                    }
                }
                if(null != latencyPayload){
                    payload.addField(ClientTopicKeys.ClientAfterFix, System.currentTimeMillis());
                }
            }
            else
            {
            	s_logger.info("ClientMediator Msg Dropped fixMessage is null:" + payload.getParsedMessage());
            }
        }
        
        

        public void onParsedMessage(IParsedPayload payload)
        {
            if(null == payload)
            {
                return;
            }
            
//            s_logger.info("ClientMediator onParsedMessage payload:" + payload.getParsedMessage());
             
            payload.addField(ClientTopicKeys.ClientStartTime, System.currentTimeMillis());
            if(s_logger.isDebugEnabled()){
            	s_logger.debug(new MessageText("NETLINK_0100020", payload.getParsedMessage().toString()));
            }
            if((!isSessionOpen()) && (!m_testSubscription)){
            	s_logger.info(new MessageText("NETLINK_0100009", payload.getParsedMessage().toString()));
            	return;
            }


            IParsedPayload  outMessage;
            String          messageType  = payload.getParsedStringField(ClientTopicKeys.MessageType);
            boolean         eventMessage = isEventMessage(messageType);
            if(eventMessage)
            {//events
            	outMessage = composeDisconnectEventMessage(payload);
            }
            else
            {//deals and quotes
            	outMessage = payload;
            }

            if( (outMessage != null) && (eventMessage || isValid(outMessage)) ) 
            {
            	sendFixMessage(outMessage, messageType, payload);
            	submitInternalMessage("latency",payload);
            }
        }
        
        public void convertToFixConvertorArrays(IParsedPayload payload, String list_key, 
        		String arrayKey)
        {
        	if((payload == null) || (list_key == null) || (arrayKey == null))
        	{
        		return;        		
        	}
        	
        	Map<String,Object> sourceData = payload.getParsedMessage();
        	Map<String,Object> commonData = new HashMap<String,Object>();
        	for(String nextKey : sourceData.keySet())
        	{
        		if(!nextKey.equalsIgnoreCase(list_key))
        		{
        			Object next_value = sourceData.get(nextKey);
        			if(next_value != null)
        			{
        				commonData.put(nextKey, sourceData.get(nextKey));
        			}
        		}
        	}        	
        	
        	List<Map<String,Object>> fixConvertorArray = new ArrayList<Map<String,Object>>();
        	Object payloadListObject = payload.getParsedField(list_key);
        	payload.addField(arrayKey, fixConvertorArray);
        	if(payloadListObject == null)
        	{        		
        		return;
        	}
        	List<IParsedPayload> payloadList = (List<IParsedPayload>)payloadListObject;
        	
        	int list_size = payloadList.size();
        	try
        	{
        		for(int i=0; i<list_size; i++ )
        		{
        			Map<String,Object> nextMap = new HashMap<String,Object>(payloadList.get(i).getParsedMessage());
        			Object fix_date_long = nextMap.get("fix_date_long");
        			if(fix_date_long != null)
        			{
        				if(fix_date_long instanceof Long)
        				{
        					nextMap.put("date_valuation_date", new Date((Long)fix_date_long));
        				}
        				else if(fix_date_long instanceof BigDecimal)
        				{
        					nextMap.put("date_valuation_date", new Date(((BigDecimal)fix_date_long).longValue()));
        				}
        				else if(fix_date_long instanceof String)
        				{
        					nextMap.put("date_valuation_date", new Date(Long.parseLong((String)fix_date_long)));
        				}
        			}
        			fix_date_long = nextMap.get("adjusted_timestamp");
        			if(fix_date_long != null)
        			{
        				if(fix_date_long instanceof Long)
        				{
        					nextMap.put("adjusted_timestamp_date", new Date((Long)fix_date_long));
        				}
        				else if(fix_date_long instanceof BigDecimal)
        				{
        					nextMap.put("adjusted_timestamp_date", new Date(((BigDecimal)fix_date_long).longValue()));
        				}
        				else if(fix_date_long instanceof String)
        				{
        					nextMap.put("adjusted_timestamp_date", new Date(Long.parseLong((String)fix_date_long)));
        				}
        			}        			
        			for(String nextSourceKey : commonData.keySet())
        			{
        				nextMap.put(nextSourceKey, commonData.get(nextSourceKey));
        			}
//        			nextMap.putAll(payloadList.get(i).getParsedMessage());
        			fixConvertorArray.add(nextMap);
        		}
        		payload.addField(arrayKey, fixConvertorArray);
        	}
        	catch(Exception e)
        	{
        		s_logger.error("ClientMediator convertToFixConvertorArrays Exception:", e);
        	}
        	
        }
        
        protected void addSecondSizesToSnapshotEntries(List<Map<String,Object>> listOfEntries) {
        	for ( Map<String,Object> currSnapshotEntryMap : listOfEntries ) {
        		addTotalAmountAsSecondSize  (currSnapshotEntryMap);
        	}
        }
        
        public void convertToFixFromDBArrays(IParsedPayload payload, String arrayKey)
        {
        	try
        	{
        		if((payload == null) || (arrayKey == null))
        		{
        			return;        		
        		}
        	
        		List<Map<String,Object>> destinationList = new ArrayList<Map<String,Object>>();
        		payload.addField(arrayKey, destinationList);        	
        	
        		int ob_size = payload.getParsedIntField("ob_size", 0);
 	
        		if(ob_size <= 0)
        		{
        			return;
        		}
        	        	        	
        		for(int i=0; i<ob_size; i++)
        		{
        			HashMap<String,Object> nextMap = new HashMap<String,Object>();
        			destinationList.add(nextMap);
        			nextMap.put("qte_type", payload.getParsedField("qte_type")); 
        			
        			String[] nextStringArray = null;
        			
        			Object nextObjectArray = payload.getParsedField("ebs_trans_id_array");        			
        			if(nextObjectArray != null)
        			{
        				nextStringArray = (String[])nextObjectArray;
        				if(nextStringArray.length > i)
            			{
        					if(nextStringArray[i] != null)
        					{
        						nextMap.put("ebs_trans_id", nextStringArray[i]);
        					}        					
            			}
        				else
            			{
            				s_logger.error("ClientMediator convertToFixFromDBArrays can not retrieve from ebs_trans_id_array i=" + i);
            			}
        			}
        			else
        			{
        				s_logger.error("ClientMediator convertToFixFromDBArrays ebs_trans_id_array is null");
        			}
        			
        			
        			nextObjectArray = payload.getParsedField("prc_qte_value_array");
        			if(nextObjectArray != null)
        			{
        				nextStringArray = (String[])nextObjectArray;
        				if(nextStringArray.length > i)
            			{
        					if(nextStringArray[i] != null)
        					{
        						nextMap.put("prc_qte_value", nextStringArray[i]);
        					}        					
            			}
        				else
            			{
            				s_logger.error("ClientMediator convertToFixFromDBArrays can not retrieve from prc_qte_value_array i=" + i);
            			}
        			}
        			else
        			{
        				s_logger.error("ClientMediator convertToFixFromDBArrays prc_qte_value_array is null");
        			}
        			
        			
        			BigDecimal[] nextBigDecimalArray = null;
        			
        			extractArrayElement_long(payload, i, nextMap, "qte_amt_avail_result_array",   "qte_amt_avail_result");
        			        
        			extractArrayElement_long(payload, i, nextMap, "qte_amt_result_array",         "qte_amt_result");

        			extractArrayElement_long(payload, i, nextMap, "qte_amt_display_result_array", "qte_amt_display_result");
        			// reading boolean field is_iceberg, so values would be 0 or 1.
        			extractArrayElement_long(payload, i, nextMap, "is_iceberg_array", "is_iceberg");
        			
        			nextObjectArray = payload.getParsedField("adjusted_timestamp_array");        			
        			if(nextObjectArray != null)
        			{
        				nextBigDecimalArray = (BigDecimal[])nextObjectArray;
        				if(nextBigDecimalArray.length > i)
            			{
        					if(nextBigDecimalArray[i] != null)
        					{
        						nextMap.put("adjusted_timestamp", nextBigDecimalArray[i].longValue());
        						nextMap.put("adjusted_timestamp_date", new Date(nextBigDecimalArray[i].longValue()));
        					}        					
            			}
        				else
            			{
            				s_logger.error("ClientMediator convertToFixFromDBArrays can not retrieve from adjusted_timestamp_array i=" + i);
            			}
        			}
        			else
        			{
        				s_logger.error("ClientMediator convertToFixFromDBArrays adjusted_timestamp_array is null");
        			}

        			
        			
        			nextMap.put("arb_date", payload.getParsedField("arb_date"));
        			
        			
        			nextObjectArray = payload.getParsedField("qte_floor_code_array");
        			if(nextObjectArray != null)
        			{
        				nextStringArray = (String[])nextObjectArray;
        				if(nextStringArray.length > i)
            			{
        					if(nextStringArray[i] != null)
        					{
        						nextMap.put("qte_floor_code", nextStringArray[i]);
        					}        					
            			}
        				else
            			{
            				s_logger.error("ClientMediator convertToFixFromDBArrays can not retrieve from qte_floor_code_array i=" + i);
            			}
        			}
        			else
        			{
        				s_logger.error("ClientMediator convertToFixFromDBArrays qte_floor_code_array is null");
        			}
        			        			
        			nextObjectArray = payload.getParsedField("arb_region_array");
        			if(nextObjectArray != null)
        			{
        				nextStringArray = (String[])nextObjectArray;
        				if(nextStringArray.length > i)
            			{
        					if(nextStringArray[i] != null)
        					{
        						nextMap.put("arb_region", nextStringArray[i]);
        					}        					
            			}
        				else
            			{
            				s_logger.error("ClientMediator convertToFixFromDBArrays can not retrieve from arb_region_array i=" + i);
            			}
        			}
        			else
        			{
        				s_logger.error("ClientMediator convertToFixFromDBArrays arb_region_array is null");
        			}
        			
        			
        			nextObjectArray = payload.getParsedField("trader_id_array");
        			if(nextObjectArray != null)
        			{
        				nextStringArray = (String[])nextObjectArray;
        				if(nextStringArray.length > i)
            			{
        					if(nextStringArray[i] != null)
        					{
        						nextMap.put("trader_id", nextStringArray[i]);
        					}        					
            			}
        				else
            			{
            				s_logger.error("ClientMediator convertToFixFromDBArrays can not retrieve from trader_id_array i=" + i);
            			}

        			}
        			else
        			{
        				s_logger.error("ClientMediator convertToFixFromDBArrays trader_id_array is null");
        			}
        			        			
        			nextObjectArray = payload.getParsedField("fix_id_array");        			
        			if(nextObjectArray != null)
        			{
        				nextBigDecimalArray = (BigDecimal[])nextObjectArray;
        				if(nextBigDecimalArray.length > i)
            			{
        					if(nextBigDecimalArray[i] != null)
        					{
        						nextMap.put("fix_id", nextBigDecimalArray[i].longValue());
        					}        					
            			}
        				else
            			{
            				s_logger.error("ClientMediator convertToFixFromDBArrays can not retrieve from fix_id_array i=" + i);
            			}
        			}
        			else
        			{
        				s_logger.error("ClientMediator convertToFixFromDBArrays fix_id_array is null");
        			}
        			
        			nextObjectArray = payload.getParsedField("fix_date_long_array");        			
        			if(nextObjectArray != null)
        			{
        				nextBigDecimalArray = (BigDecimal[])nextObjectArray;
        				if(nextBigDecimalArray.length > i)
            			{
        					if(nextBigDecimalArray[i] != null)
        					{
        						long fix_date_long = nextBigDecimalArray[i].longValue();
        						nextMap.put("fix_date_long", fix_date_long);
        						nextMap.put("date_valuation_date", new Date(fix_date_long));
        					}        					
            			}
        				else
            			{
            				s_logger.error("ClientMediator convertToFixFromDBArrays can not retrieve from fix_date_long_array i=" + i);
            			}
        			}
        			else
        			{
        				s_logger.error("ClientMediator convertToFixFromDBArrays fix_date_long_array is null");
        			}
        		}
        	}
        	catch(Exception e)
        	{
        		s_logger.error("unexpected exception", e);
        	}
        }


		protected void extractArrayElement_long(IParsedPayload payload, int ix, HashMap<String, Object> theMap, String fieldNameOfArray, String fieldNameOfSingle) {
			Object nextObjectArray;
			BigDecimal[] nextBigDecimalArray;
			nextObjectArray = payload.getParsedField(fieldNameOfArray);
			if(nextObjectArray != null)
			{
				nextBigDecimalArray = (BigDecimal[])nextObjectArray;
				if(nextBigDecimalArray.length > ix)
				{
					if(nextBigDecimalArray[ix] != null)
					{
						theMap.put(fieldNameOfSingle, nextBigDecimalArray[ix].longValue());
					}        					
				}
				else
				{
					s_logger.error(String.format("ClientMediator convertToFixFromDBArrays can not retrieve from '%s', i='%d'", fieldNameOfArray, ix));
				}
			}
			else
			{
				s_logger.error(String.format("ClientMediator convertToFixFromDBArrays '%s' is null", fieldNameOfArray));
			}
		}        
                
        
        /**
         * From MessageListener
         * @param message Message - message from the router
         */
        public void onMessage(Message message) {
        	synchronized(m_monitor){        		        	
            try {
                IParsedPayload payload = (IParsedPayload) ( (ObjectMessage) message).getObject();
                
                if(null == payload){
                    s_logger.error(new MessageText("NETLINK_0100028", message));
                    return;
                }
                payload.addField(ClientTopicKeys.ClientStartTime, System.currentTimeMillis());
                if(s_logger.isDebugEnabled()){
                    s_logger.debug(new MessageText("NETLINK_0100020", payload.getParsedMessage().toString()));
                }
                if((!isSessionOpen()) && (!m_testSubscription)){
                    s_logger.info(new MessageText("NETLINK_0100009", payload.getParsedMessage().toString()));
                    return;
                }

//                s_logger.info("ClientMediator received msg:" + payload.getParsedMessage());
                
                
                //check subscription id
//                int subscriptionId = payload.getParsedIntField(ClientTopicKeys.SubscriptionId);
//                if(m_currentSubscriptionId != subscriptionId){
//                    s_logger.info(new MessageText("NETLINK_0100008", subscriptionId, m_currentSubscriptionId));
//                    return;
//                }

                IParsedPayload  outMessage;
                final int msgVersion = makeMinimumMessageVersion(payload, 2);
                String messageType  = payload.getParsedStringField("mq_msg_type");
                String instrument = payload.getParsedStringField("instrument","");
                if(messageType.equalsIgnoreCase("RTCK"))
                {
                	return;
                }
                // make sure on payload it's numeric
                Integer matchingType = payload.getParsedIntField("matching_type", MT_LIMIT);
                payload.addField("matching_type", matchingType);
                
                
                if(instrument.equalsIgnoreCase("NDF"))
                {
                	if((matchingType == 1)||(matchingType == 5))
                	{
                		s_logger.info("ClientMediator dropped to the floor invalid instrument NDF with matching type " + 
                				matchingType + " as invalid message type. msg:" + payload.getParsedMessage());
                		return;
                	}
                }                
                else if(instrument.equalsIgnoreCase("SPOT"))
                {
                	if((matchingType == 3)||(matchingType == 4)||(matchingType == 5))
                	{
                		s_logger.info("ClientMediator dropped to the floor invalid instrument SPOT with matching type " + 
                				matchingType + " as invalid message type. msg:" + payload.getParsedMessage());
                		return;
                	}
                }
                else if(instrument.equalsIgnoreCase("FWD"))
                {
                	if((matchingType == 0)||(matchingType == 3)||(matchingType == 4))
                	{
                		s_logger.info("ClientMediator dropped to the floor invalid instrument FWD with matching type " + 
                				matchingType + " as invalid message type. msg:" + payload.getParsedMessage());
                		return;
                	}
                }

                
                handleTradeType(payload, matchingType);
                
                String base_ccy = payload.getParsedStringField("base_ccy","");
                String local_ccy = payload.getParsedStringField("local_ccy","");
                String short_alias_name = m_pair_names.get(base_ccy + local_ccy);
                String pair_id = m_pair_name_to_id.get(base_ccy + local_ccy);
                payload.addField("ccy_pair_name", short_alias_name);
                
                String tenor = payload.getParsedStringField("tenor","");
                String near_tenor = payload.getParsedStringField("near_tenor", "");
                if(near_tenor.length() <= 0)
                {
                	payload.getParsedMessage().remove("near_tenor");
                }
                String qte_type_name = payload.getParsedStringField("qte_type_name","_");
                String market_instrument = base_ccy + ":" + local_ccy + ":" + 
                                           instrument + ":" + tenor + ":" + qte_type_name;
                if(tenor.equalsIgnoreCase("SPOT"))
                {
                	payload.addField("tenor", "0");
                }
                long msg_timestamp_long = payload.getParsedLongField("msg_timestamp_long",0L);
                Date arb_date = new Date(msg_timestamp_long);
                payload.addField("arb_date", arb_date);
                // make sure to get qte_type as int and convert it to string
                int qte_type = payload.getParsedIntField("qte_type", 0);
                payload.addField("qte_type", "" + qte_type);
                
                if(messageType.equalsIgnoreCase("ROBM"))
                {
                	payload.addField("msg_sub_type", "M0056");                	
                	if(!m_snapshot_filter.containsKey(market_instrument))
                	{
                		m_snapshot_filter.put(market_instrument, qte_type_name);
                	}                	                
                	
                	// should be MktQteInfo4 or MktQteInfo5 or any MktQteInfoX
                	final String nameOfEntryFromMq = "MktQteInfo" + msgVersion;
                	
                	Object mkt_qte_info = payload.getParsedField(nameOfEntryFromMq);
                	
                	if(mkt_qte_info == null)
                	{
                		convertToFixFromDBArrays(payload, "mkt_qte_info_element");
                	}
                	else
                	{
                		convertToFixConvertorArrays(payload, nameOfEntryFromMq, "mkt_qte_info_element");
                	}
//                	s_logger.info("ClientMediator array for the translator is created. payload:" + payload.getParsedMessage());
                	
                	// adding second sizes
                	List<Map<String,Object>> listOfSnapshotEntries = (List<Map<String,Object>>)
                		payload.getParsedField("mkt_qte_info_element");
                	
                	addSecondSizesToSnapshotEntries(listOfSnapshotEntries);
                	
                }
                
                Object fix_id_object = payload.getParsedField("fix_id");
                
                long fix_id = 0L;
                
                if((fix_id_object != null)&&(fix_id_object instanceof Long))
                {
                	fix_id = (Long)fix_id_object;
                }
                else if((fix_id_object != null)&&(fix_id_object instanceof String))
                {
                	fix_id = Long.parseLong((String)fix_id_object);
                }
                
                if(fix_id > 0L)
                {
                	Object fixing_date_long_object = payload.getParsedField("fix_date_long");
                	long fixing_date_long = 0L;
                	
                	if(fixing_date_long_object == null)
                	{
                		fixing_date_long = 0L;
                	}
                	else if(fixing_date_long_object instanceof Long)
                	{
                		fixing_date_long = (Long)fixing_date_long_object;
                	}
                	else if(fixing_date_long_object instanceof String)
                	{
                		fixing_date_long = Long.parseLong((String)fixing_date_long_object);
                	}
                	
                	if(fixing_date_long > 0)
                	{
                		payload.addField("date_valuation_date", new Date(fixing_date_long));
                	}
                	
                	if(m_fix_names.containsKey("" + fix_id))
                	{
                		payload.addField("fixing_source", m_fix_names.get("" + fix_id));
                	}
                }

                String operation = "0";
                boolean toAddDeltaAsSecondSize = false;
                
                if(messageType.equalsIgnoreCase("RDST"))
                {
                	Object deal_amt_result = payload.getParsedField("deal_amt_result");
                	long deal_amt = 0L;
                	if(deal_amt_result instanceof Long)
                	{
                		deal_amt = (Long)deal_amt_result;
                	}
                	else if (deal_amt_result instanceof String)
                	{
                		deal_amt = Long.parseLong((String)deal_amt_result);
                	}
                	else if(deal_amt_result instanceof Integer)
                	{
                		deal_amt = ((Integer)deal_amt_result).longValue();
                	}
                	
                	payload.addField("deal_amt_result", deal_amt);
                	
                	if(deal_amt <= 0)
                	{
                		s_logger.info("ClientMediator Deal message is filtered out. Amount is 0. payload:" + payload.getParsedMessage());
                		return;
                	}
                	
                	deal_amt_result = payload.getParsedField("order_ref");
                	
                	if((deal_amt_result != null)&&(((String)deal_amt_result).length() <= 0))
                	{
                		payload.addField("order_ref", " ");
                	}
                	
                	payload.addField("msg_sub_type", "M0052");
                	payload.addField("deal_completion_time", arb_date);
                	long date_long = payload.getParsedLongField("valuation_date_long",0L);
                	if(fix_id <= 0)
                	{                		
                		if(date_long > 0)
                		{
                			payload.addField("date_valuation_date", new Date(date_long));
                		}                		
                	}
                	date_long = payload.getParsedLongField("settlement_date_long",0L);
                	if(date_long > 0)
            		{
            			payload.addField("date_settlement_date", new Date(date_long));
            		}
                	date_long = payload.getParsedLongField("value_date_long",0L);
                	if(date_long > 0)
            		{
            			payload.addField("date_value_date", new Date(date_long));
            		} 
                	date_long = payload.getParsedLongField("trade_date_long",0L);
                	if(date_long > 0)
            		{
            			payload.addField("date_trade_date", new Date(date_long));
            		}
                	
                	int deal_side = 0;
                	Object deal_side_obj = payload.getParsedField("deal_side");
                	if(deal_side_obj == null)
                	{
                		deal_side = 0;
                	}
                	else if(deal_side_obj instanceof Integer)
                	{
                		deal_side = (Integer)deal_side_obj;
                	}
                	else if(deal_side_obj instanceof String)
                	{
                		deal_side = Integer.parseInt((String)deal_side_obj);
                	}
                	payload.addField("deal_side", deal_side + 1);
                	if(instrument.equalsIgnoreCase("SPOT"))
                	{
                		payload.addField("date_settlement_date", payload.getParsedField("date_value_date"));
                	}
                	
                	String eff_price = payload.getParsedStringField("effective_deal_price_value", "0.0"); 
                	if(eff_price.trim().length() <= 0)
                	{
                		eff_price = "0.0";
                		payload.addField("effective_deal_price_value", eff_price);
                	}
                }
                else if(messageType.equalsIgnoreCase("NFRT"))
                {
                	payload.addField("msg_sub_type", "M0058");
                	operation = "0";
                	payload.addField("operation", operation);
                	payload.addField("qte_type", "W");
                	payload.addField("ebs_trans_id", "" + fix_id);
                	payload.addField("date_valuation_date", new Date(payload.getParsedLongField("fix_date_long", 0L)));
                	if(pair_id != null)
                	{
                		Map<String,Object> nextPair = m_pairs.get(pair_id);
                		if(nextPair.containsKey("instrument"))
                		{
                			payload.addField("instrument", nextPair.get("instrument"));
                		}
                		else
                		{
                			payload.addField("instrument", "SPOT");
                		}
                		if(nextPair.containsKey("tenor"))
                		{
                			payload.addField("tenor", nextPair.get("tenor"));
                		}
                		else
                		{
                			payload.addField("tenor", "SPOT");
                		}
                	}
                	else
                	{
                		payload.addField("instrument", "SPOT");
                		payload.addField("tenor", "SPOT");
                	}
                }
                else
                {
                	if(!m_snapshot_filter.containsKey(market_instrument))
                	{
                		s_logger.info("ClientMediator Msg dropped filtered by market_instrument:" + market_instrument +
                				      " msg:" + payload.getParsedMessage() + 
                				      " available instruments:" + m_snapshot_filter.keySet());
                		return;
                	}                	
                }
                
                payload.addField("qte_upd_num_rep_groups", 1); 
                
                
                if(messageType.equalsIgnoreCase("RQTN"))
                {
                	operation = "2";
                	payload.addField("operation", operation);
                	payload.addField("msg_sub_type", "M0058");
                }
                
                if(messageType.equalsIgnoreCase("RMQI"))
                {
                	operation = "2";
                	payload.addField("operation", operation);
                	payload.addField("msg_sub_type", "M0058");
                }

                if(messageType.equalsIgnoreCase("RQTR"))
                {
                	operation = "D"; // "D" should be deprecated, use "1" instead
                	payload.addField("operation", operation);
                	payload.addField("msg_sub_type", "M0058");
                	toAddDeltaAsSecondSize = true;
                }

                if(messageType.equalsIgnoreCase("RMAD"))
                {
                	// "D" is custom value for Decrease. consider removing it and use only "1" operation which means change
                	operation = "D";

                	/*
                	   for BackWard compatibility, will not send operation "1"
                	int amendType = payload.getParsedIntField("amend_type");
                	// http://fixprotocol.org/FIXimate3.0/en/FIX.5.0SP2/tag279.html
                	if ( amendType >= CoreEx.CODE_MktQteAmendType_GENERICDETAILS ) {
                		// "1" = "change"
                		operation = "1";
                	}
                	*/
                	
                	payload.addField("operation", operation);
                	payload.addField("msg_sub_type", "M0058");
                	toAddDeltaAsSecondSize = true;
                }

                if(messageType.equalsIgnoreCase("RQTU"))
                {
                	operation = "I"; // "I" should be deprecated, use "1" instead
                	payload.addField("operation", operation);
                	payload.addField("msg_sub_type", "M0058");
                	toAddDeltaAsSecondSize = true;
                }

                if(messageType.equalsIgnoreCase("RQTM"))
                {
                	operation = "0";
                	payload.addField("operation", operation);
                	payload.addField("msg_sub_type", "M0058");
                	
                	addTotalAmountAsSecondSize(payload);
                }
                
                if ( toAddDeltaAsSecondSize ) {
                	addDeltaAmountAsSecondSize(payload);
                }
                
                boolean         eventMessage = isEventMessage(messageType);
                
                if(eventMessage){//events
                    outMessage = composeDisconnectEventMessage(payload);
                }else{//deals and quotes
                	s_logger.info("ClientMediator received valid msg:" + payload.getParsedMessage());
                        outMessage = payload;
                }

                if( (outMessage != null) && (eventMessage || isValid(outMessage)) ) {
                    sendFixMessage(outMessage, messageType, payload);
//                    m_destinations.sendMessage(Keys.MessageTypes.latency_messages, payload);
                    submitInternalMessage("latency_msg",payload);
                }
                else
                {
                	s_logger.info("ClientMediator Not valid msg drooped:" + outMessage);
                }
            }
            catch (JMSException ex) {
                s_logger.error(new MessageText("NETLINK_0100004", message), ex);
            }
            catch(Exception e)
            {
            	s_logger.error("ClientMediator onMessage Exception:",e);
            }
        }
        }

		protected void handleTradeType(IParsedPayload payload, Integer matchingType) {
			int tradeTypeToPut = TRADE_TYPE_SPOT;
            int fieldToCheck   = matchingType;

            int  existingFieldTradeType = payload.getParsedIntField("trade_type", -1);
			if ( existingFieldTradeType >= 0) {
				fieldToCheck = existingFieldTradeType;
			}

			switch (fieldToCheck) {
			case MT_LIMIT:
				tradeTypeToPut = TRADE_TYPE_SPOT;
				break;
			case MT_FIX: // also equals to TRADE_TYPE_FIX
			case MT_NDF_SL:
			case MT_NDF_ML:
			case MT_SWAPS:
				tradeTypeToPut = TRADE_TYPE_FIX;
				break;
			case MT_CM:
			case TRADE_TYPE_CM:
				tradeTypeToPut = TRADE_TYPE_CM;
				break;
			default:
				// if nothing special was found than default value will be SPOT
				tradeTypeToPut = TRADE_TYPE_SPOT;
			}
				
			payload.addField("trade_type", tradeTypeToPut);
		}

        private static boolean isEventMessage(String messageType) {
            return messageType.equalsIgnoreCase("EVNT");
        }

        private boolean isValid(IParsedPayload message) {
            boolean valid       = null != message;
            return valid;
        }

        private IParsedPayload composeDisconnectEventMessage(IParsedPayload routerMessage){
            IParsedPayload result = new ParsedPayload();
            int event_code = routerMessage.getParsedShortField(ClientTopicKeys.EventCode);
            int statusID = 1;
            String status = "";
            switch(event_code){
                case EVENT_ARB_DOWN:
                    statusID = 2;
                    status = "Arb Down";
                    break;
                case EVENT_ARB_UP_LIVE:
                case EVENT_ARB_UP_MOCK:
                    statusID = 1;
                    status = "Arb Up";
                    break;
                case EVENT_CONNECTED:
                    statusID = 1;
                    status = "Upstream Connection Up";
                    break;
                case EVENT_DOWN:
                    statusID = 2;
                    status = "Upstream Connection Down";
                    break;
                default:
                    if(s_logger.isDebugEnabled()){
                        s_logger.debug(new MessageText("NETLINK_0100021", event_code));
                    }
                    return null;
            }

            result.addField(ClientTopicKeys.NetworkResponseId, m_disconnectEventCounter++);
            result.addField(ClientTopicKeys.ConnectionStatusId, statusID);
            result.addField(ClientTopicKeys.ConnectionStatusText, status);
            if(s_logger.isDebugEnabled()){
                s_logger.debug("******  Event message: " + result.toString());
            }
            return result;
        }

        private void forceClientLogout(String errorCode, String reason, boolean sendJmx){
            forceClientLogout(new MessageText(errorCode), reason, sendJmx);
        }

        private void forceClientLogout(MessageText mess, String reason, boolean sendJmx){
            s_logger.error(mess);
            if(null == reason){
                reason = mess.getText();
            }
            synchronized(m_monitor){
                onClientLogout(reason, false);
            }
            sendLogoutMessages(mess, reason, sendJmx);
        }

        /**
         * Send logout Fix message and JMX alert
         * @param mess MessageText
         * @param reason String
         * @param sendJmx boolean
         */
        private void sendLogoutMessages(MessageText mess, String reason, boolean sendJmx) {
            if(null == reason){
                reason = mess.getText();
            }
            m_fixSession.logoutClient(reason);
            if(sendJmx){
                IParsedPayload payload = new ParsedPayload();
                payload.addField("type", -2);
                payload.addField(JmxTopicKeys.NotificationType, JmxTopicKeys.NotificationTypes.Error);
                payload.addField(JmxTopicKeys.Text, mess.getText());
//                m_destinations.sendMessage(Keys.MessageTypes.jmx_messages, payload);
                submitInternalMessage("jmxAlerts",payload);
            }
        }

        
    	private void submitInternalMessage(String topicName, IParsedPayload payload)
    	{
    		if(m_NdsChannel == null)
        	{
    			return;
        	}
    		try
    		{
    			m_NdsChannel.submitMessage(topicName, payload);
    		}
    		catch(JMSException jmse)
    		{
    			s_logger.error("ClientMediator JMSException:",jmse);
    		}
    	}
        
        
        State getState(){
            synchronized(m_monitor){
                return m_state;
            }
        }
        
        
    	public class PeriodicRecovery implements Runnable
    	{
    	    public PeriodicRecovery() { }


    	    public void run()
    	    {
    	            try
    	            {
    	            	refreshMaps();
    	            }
    	            catch (Exception ex)
    	            {
    	            	s_logger.error(new MessageText("ClientMediator exception running refreshMaps", ex));	            
    	            }
    	    }
    	 }                
        
        
    	public class TestSubscribe implements Runnable
    	{
    	    public TestSubscribe() { }


    	    public void run()
    	    {
    	            try
    	            {
//    	            	Thread.sleep(3000);
    	            	s_logger.info("ClentMediator begin TestSubscription sequence");
    	            	m_state = State.LoggedIn;
    	            	m_bufferedMessagesCounter = 0;
    	            	IParsedPayload payload = new ParsedPayload();            	
    	                payload.addField(ClientTopicKeys.Event, 0); //connect
    	              s_logger.info(new MessageText("NETLINK_0100022", Keys.MessageTypes.client_connect, payload.getParsedMessage().toString()));
    	              submitInternalMessage("client_connection_event",payload);
    	              onClientSubscribe(payload);
    	            }
    	            catch (Exception ex)
    	            {
    	            	s_logger.error(new MessageText("ClientMediator exception running TestSubscribe", ex));	
    	            	ex.printStackTrace();
    	            }
    	    }
    	 }

    }

